package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.beans.TradeSkuOrderBean;
import com.atguigu.realtime.common.GmallConfig;
import com.atguigu.realtime.func.BeanToJsonStrMapFcuntion;
import com.atguigu.realtime.func.DimAsyncFunction;
import com.atguigu.realtime.utils.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author: 洛尘
 * @since: 2023-10-13 18:16
 * @description: sku粒度下单聚合统计
 **/
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置，开启检查点，主要原因是往doris写入的时候不开启检查点无法写入doris
        env.enableCheckpointing(5000L);
        env.setRestartStrategy(RestartStrategies.noRestart());
        //TODO 3.从kafka消费数据
        //定义消费者主题和消费者组
        String topic="dwd_trade_order_detail";
        String groupId="dws_trade_sku_order_group";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);

        SingleOutputStreamOperator<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        //TODO 4.转换数据并且过滤空消息
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                            collector.collect(jsonObj);
                        }
                    }
                }
        );
        //todo 5.去重
        //按照唯一键（订单id）进行分组
        KeyedStream<JSONObject, String> keybyDS =
                jsonDS.keyBy(jsonObject -> jsonObject.getString("id"));

        //去重方式1，状态 + 定时任务
        //缺点：每条数据进来无论是否重复，都需要等待
       /* SingleOutputStreamOperator<JSONObject> distinctDS = keybyDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> lastJsonObjState
                                = new ValueStateDescriptor<>("lastJsonObjState", JSONObject.class);

                        this.lastJsonObjState = getRuntimeContext().getState(lastJsonObjState);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
                        //从状态中获取数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj == null) {
                            //没有产生重复数据，将当前这个条数据放入状态中，并且注册一个5s后执行的定时器
                            lastJsonObjState.update(jsonObject);
                            long currentProcessingTime = context.timerService().currentProcessingTime();
                            context.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        } else {
                            //到达这里说明产生了重复数据，那么用当前数据的聚合时间和状态中的数据的聚合时间进行对比，将时间大的放入状态中
                            String ts1 = lastJsonObj.getString("聚合时间");
                            String ts2 = jsonObject.getString("聚合时间");
                            if (ts2.compareTo(ts1) >= 0) {
                                //如果当前数据的聚合时间大于等于状态中的数据的聚合时间，将当前数据放到状态中
                                lastJsonObjState.update(jsonObject);
                            }

                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        //当定时器被触发以后，将状态中的数据发送到下游
                        JSONObject jsonObject = lastJsonObjState.value();
                        out.collect(jsonObject);
                        //清除状态
                        lastJsonObjState.clear();
                    }
                }
        );*/
        //去重方式2：状态 + 抵消
        //优点：时效性号 不足：如果出现重复，会向下游发送3条数据，传输量变大
        SingleOutputStreamOperator<JSONObject> distinctDS = keybyDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> lastJsonObjState
                                = new ValueStateDescriptor<>("lastJsonObjState", JSONObject.class);
                        //因为此方法没有注册定时器，去触发定时器的功能清楚状态中的数据，所以需要设置状态失效时间
                        lastJsonObjState.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                        this.lastJsonObjState = getRuntimeContext().getState(lastJsonObjState);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject lastValueObj = lastJsonObjState.value();
                        if (lastValueObj != null) {
                            //如果状态中的数据不为空，说明出现了重复数据，那么此时需要将状态中的数据取反传递到下游
                            String splitOriginalAmount = lastValueObj.getString("split_original_amount");
                            String splitCouponAmount = lastValueObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastValueObj.getString("split_activity_amount");
                            String splitTotalAmount = lastValueObj.getString("split_total_amount");
                            lastValueObj.put("split_original_amount", "-" + splitOriginalAmount);
                            lastValueObj.put("split_coupon_amount", "-" + splitCouponAmount);
                            lastValueObj.put("split_activity_amount", "-" + splitActivityAmount);
                            lastValueObj.put("split_total_amount", "-" + splitTotalAmount);
                            collector.collect(lastValueObj);

                        }
                        //第一次运行的时候状态为空，将数据传入
                        lastJsonObjState.update(jsonObject);
                        collector.collect(jsonObject);

                    }
                }
        );
//        distinctDS.print("________");
        //todo 6.再次进行数据转换，将流中的数据jsonObject转换为实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {

                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObject) throws Exception {
                        String skuId = jsonObject.getString("sku_id");
                        String splitOriginalAmount = jsonObject.getString("split_original_amount");
                        String splitCouponAmount = jsonObject.getString("split_coupon_amount");
                        String splitActivityAmount = jsonObject.getString("split_activity_amount");
                        String splitTotalAmount = jsonObject.getString("split_total_amount");
                        Long ts = jsonObject.getLong("ts") * 1000;
                        TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                                .skuId(skuId)
                                .originalAmount(new BigDecimal(splitOriginalAmount))
                                .activityAmount(new BigDecimal(splitActivityAmount))
                                .couponAmount(new BigDecimal(splitCouponAmount))
                                .orderAmount(new BigDecimal(splitTotalAmount))
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );
        //todo 7.指定watermark以及提取的事件事件字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeSkuOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSkuOrderBean tradeSkuOrderBean, long l) {
                                        return tradeSkuOrderBean.getTs();
                                    }
                                }
                        )
        );
        //todo 8.按照统计的维度去分组sku
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS
                = withWatermarkDS.keyBy(tradeSkuOrderBean -> tradeSkuOrderBean.getSkuId());
//        skuIdKeyedDS.print("*****");
        //todo 9.统计分组以后开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
                = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));
        //todo 10.开窗聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                }
                ,
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String groupId, TimeWindow timeWindow, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                        String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());
                        String curDate = DateFormatUtil.toDate(timeWindow.getStart());
                        for (TradeSkuOrderBean tradeSkuOrderBean : iterable) {
                            tradeSkuOrderBean.setStt(stt);
                            tradeSkuOrderBean.setEdt(edt);
                            tradeSkuOrderBean.setCurDate(curDate);
                            collector.collect(tradeSkuOrderBean);
                        }
                    }
                }
        );
//        reduceDS.print("!!!!");
        //TODO 11.关联sku维度
        //基本维度关联实现
        /*final SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private Connection conn;
                    private Jedis jedis;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = HbaseUtil.getHbaseConnection();
                        jedis = RedisUtil.getRedisClient();
                    }

                    @Override
                    public void close() throws Exception {
                        HbaseUtil.closeHbaseConnection(conn);
                        RedisUtil.closeRedisClient(jedis);

                    }
                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                        //根据流中的对象获取要关联的主键
                        String skuId = tradeSkuOrderBean.getSkuId();
                        //根据维度的主键获取维度对象
                        JSONObject dimJsonObj = DimUtil.getDimInfo(jedis, conn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                        //将维度对象相关属性补充到流中的对象身上
                        tradeSkuOrderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        tradeSkuOrderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        tradeSkuOrderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        tradeSkuOrderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        return tradeSkuOrderBean;
                    }
                }
        );
         */
//        withSkuInfoDS.print(">>>>");
         SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                    @Override
                    public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));

                    }
                    @Override
                    public String getKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                },
                60, TimeUnit.SECONDS
        );

        //TODO 12.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
                withSkuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                    @Override
                    public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                    }

                    @Override
                    public String getKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSpuId();
                    }
                },
                60, TimeUnit.SECONDS
        );


        //TODO 13.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
                withSpuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                    @Override
                    public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                    }

                    @Override
                    public String getKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getTrademarkId();
                    }
                },
                60, TimeUnit.SECONDS
        );
        // withTmDS.print(">>>");
        //TODO 14.关联categor3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
                withTmDS,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {

                    @Override
                    public String getKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public void join(TradeSkuOrderBean bean,
                                     JSONObject dim) {
                        bean.setCategory3Name(dim.getString("name"));
                        bean.setCategory2Id(dim.getString("category2_id"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 15.关联categor2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
                c3Stream,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {

                    @Override
                    public String getKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public void join(TradeSkuOrderBean bean,
                                     JSONObject dim) {
                        bean.setCategory2Name(dim.getString("name"));
                        bean.setCategory1Id(dim.getString("category1_id"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 16.关联categor1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> restulStream = AsyncDataStream.unorderedWait(
                c2Stream,
                new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {

                    @Override
                    public String getKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public void join(TradeSkuOrderBean bean,
                                     JSONObject dim) {
                        bean.setCategory1Name(dim.getString("name"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        restulStream.print(">>>");
        //TODO 17.将关联的结果写到Doris中
        restulStream
                .map(new BeanToJsonStrMapFcuntion<>())
                .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));

        env.execute();
    }
}